package com.atguigu.transform;

import com.alibaba.fastjson.JSON;
import com.atguigu.func.ClickSource;
import com.atguigu.pojo.Event;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.checkerframework.checker.units.qual.C;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/14 18:30
 *
 * process算子：
 *      配合处理函数使用。
 *
 *      处理函数的功能：
 *          1. 拥有富函数提供的生命周期方法 以及 可以获取到运行时上下文，进行状态编程
 *
 *          2. 可以获取到TimerService,可以注册定时器，在计划性的未来的某个时间做一些处理
 *              当闹钟响了，可以自动调用onTimer方法，我们可以将处理逻辑写到onTimer方法中。
 *
 *          3. 可以使用侧输出流
 *
 *
 */
public class Flink07_Process {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);
        //获取数据
        DataStreamSource<Event> ds = env.addSource(new ClickSource());

        ds.process(
                new ProcessFunction<Event, String>() {

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                    }

                    @Override
                    public void processElement(Event value, ProcessFunction<Event, String>.Context context, Collector<String> out) throws Exception {

                        out.collect(JSON.toJSONString(value));

                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                    }
                }
        ).print();


        env.execute();

    }
}
